package com.moli.iov.handle.alarm.mq.kafka;


import com.moli.iov.async.BlockingQueueThreadPool;
import com.moli.iov.consumer.KafkaBatchConsumerServer;
import com.moli.iov.handle.alarm.biz.RealHandler;
import com.moli.iov.protocol.codec.DataDecoder;
import com.moli.iov.protocol.codec.ReqStreamData;
import com.moli.iov.protocol.model.realtime.RealTimeInfoReq;
import java.util.List;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.Acknowledgment;

/**
 * <p>
 *  实时数据报警  批量拉取kafka消息
 * </p>
 *
 * @author: zql
 * @Date: 2019/2/13 11:08
 */
@Slf4j
public class RealBatchConsumerListener extends KafkaBatchConsumerServer {

    @Resource(name = "realBlockingQueueThreadPool")
    private BlockingQueueThreadPool blockingQueueThreadPool;

    @Autowired
    private RealHandler realHandler;

    @Override
    public void manualHandle(List<ConsumerRecord<String, Object>> consumerRecords, Acknowledgment acknowledgment) {
        try{
            for(ConsumerRecord<String, Object> record : consumerRecords){
                RealTimeInfoReq req = decode(record);
                if(req == null){
                    continue;
                }
                realHandle(record, req);
            }
        }catch(Exception e){
            log.error("kafka 消费者  实时报警业处理  异常", e);
        }finally{
            acknowledgment.acknowledge();
        }
    }

    /**
     * 实时上数据 报警
     * @param realTimeInfoReq
     */
    private void realHandle(ConsumerRecord<String, Object> record, RealTimeInfoReq realTimeInfoReq) {
        try{
            if(realTimeInfoReq == null || realTimeInfoReq.getAlarmReq() == null){
                log.error("Kafka 消费者  实时报警业处理 Batch 无报警信息 topic: {} 数据：{}  offset:{} partition:{}",
                        record.topic(), realTimeInfoReq, record.offset(), record.partition());
                return;
            }

            blockingQueueThreadPool.execute(realTimeInfoReq.vin, () -> realHandler.handle(realTimeInfoReq));
        }catch(Exception e){
            log.error("Kafka 消费者 realHandle 异常", e);
        }

    }

    /**
     * 解码为 RealTimeInfoReq
     * @param record
     * @return
     */
    private RealTimeInfoReq decode(ConsumerRecord<String, Object> record){
        try {
            ReqStreamData stream = (ReqStreamData) record.value();
            DataDecoder<RealTimeInfoReq> decoder = new DataDecoder<>();
            RealTimeInfoReq req = new RealTimeInfoReq();
            decoder.decode(req, stream.getDataBytes());
            return req;
        } catch (Exception e) {
            log.error("解码为 RealTimeInfoReq 异常", e);
            return null;
        }
    }

}
